
package com.lee.service; 
 
import com.alibaba.fastjson.JSON; 
import com.alibaba.fastjson.JSONObject; 
import com.lee.info.UserFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.source.HbaseSource; 
import org.apache.flink.api.common.serialization.SimpleStringSchema; 
import org.apache.flink.streaming.api.datastream.DataStreamSource; 
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; 
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; 
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; 
 
import java.util.Properties; 
 
public class FlinkStream { 
    public static void main(String[] args) throws Exception {
        //从kafka获取数据
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers","bds1:9092,bds2:9092");
        properties.setProperty("group.id","consumer-group");
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset","latest");
 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        FlinkKafkaConsumer<String> test = new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties);
        DataStreamSource<String> source = env.addSource(test);
        //打印获取的数据
//        source.print();
        //将json数据转换为对象
        SingleOutputStreamOperator<UserFlowInfo> dataStream = source.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            UserFlowInfo userFlowInfo = new UserFlowInfo();
            userFlowInfo.setCustId(String.valueOf(jsonObject.get("cust_id")));
            userFlowInfo.setName(String.valueOf(jsonObject.get("name")));
            userFlowInfo.setActivityCode(String.valueOf(jsonObject.get("activity_code")));
            userFlowInfo.setActivityName(String.valueOf(jsonObject.get("activity_name")));
            userFlowInfo.setTime(String.valueOf(jsonObject.get("time")));
            return userFlowInfo;
        });
        //打印每个用户的所有信息
        dataStream.map(item->{
            String custId = item.getCustId();
            String crowdInfo = HbaseSource.getCrowdInfo(custId);
            Logger.info(crowdInfo);
            return crowdInfo;
        });
 
        env.execute();
 
    }
} 
